package net.mysgame.czfun.store.redis;

import org.apache.commons.beanutils.BeanUtilsBean;
import org.apache.commons.beanutils.BeanUtilsBean2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cglib.beans.BeanMap;
import org.springframework.util.CollectionUtils;
import org.springframework.util.SerializationUtils;
import redis.clients.jedis.*;
import redis.clients.jedis.exceptions.JedisException;
import redis.clients.util.SafeEncoder;
import sun.plugin.com.TypeConverter;

import javax.annotation.Resource;
import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Modifier;
import java.math.BigDecimal;
import java.util.*;

/**
 * Created by manatea on 2016/5/28.
 */
public class RedisClientTemplate implements ICacheProtocol {

    private static final Logger LOGGER = LoggerFactory.getLogger(RedisClientTemplate.class);

    @Resource(name = "redisDataSource")
    protected RedisDataSource redisDataSource;

    public void disconnect() {
        ShardedJedis shardedJedis = redisDataSource.getRedisClient();
        shardedJedis.disconnect();
    }

    private BeanUtilsBean beanUtilsBean = BeanUtilsBean2.getInstance();

    private static int MAX_CACHE_OPERATE_COUNT = 3;

    public List<Long> convertStringCollectionToLongList(Collection<String> strings) {
        List<Long> longs = new ArrayList<Long>();
        for (String string : strings) {
            try {
                longs.add(Long.parseLong(string));
            } catch (Exception e) {
                LOGGER.warn("failed to parse long, string=" + string, e);
            }
        }
        return longs;
    }

    public List<String> convertLongCollectionToStringList(Collection<Long> longs) {
        List<String> strings = new ArrayList<String>();
        for (long l : longs) {
            try {
                strings.add(String.valueOf(l));
            } catch (Exception e) {
                LOGGER.warn("failed to parse long, long=" + l, e);
            }
        }
        return strings;
    }

    /**
     * transaction cache operation
     *
     * @param key
     * @throws java.io.IOException
     */
    public void transaction(String key, CacheOperationFactory factory) throws IOException {
        boolean isFailed = false;
        ShardedJedis jedis = redisDataSource.getRedisClient();
        try {
            Jedis shard = jedis.getShard(SafeEncoder.encode(key));
            shard.watch(SafeEncoder.encode(key));
            Transaction t = shard.multi();
            factory.setJedis(shard);
            factory.cache();
            t.exec();
        } catch (JedisException e) {
            isFailed = true;
            throw e;
        } finally {
            redisDataSource.returnResource(jedis, isFailed);
        }
    }

    protected abstract class CacheOperationFactory {
        private Jedis jedis;

        public void setJedis(Jedis jedis) {
            this.jedis = jedis;
        }

        public Jedis getJedis() {
            return this.jedis;
        }

        public abstract void cache() throws JedisException, IOException;
    }

    public String set(final String key, final Object o) throws IOException {
        return new CountOperationFactory<String>(MAX_CACHE_OPERATE_COUNT, key) {
            public String cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                String result = jedis.set(SafeEncoder.encode(key), SerializationUtils.serialize(o));
                return result;
            }
        }.operate();
    }

    public List<String> set(final String[] keys, final Object[] os) throws IOException {
        return new CountOperationFactory<List<String>>(MAX_CACHE_OPERATE_COUNT, keys) {
            public List<String> cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                List<String> ts = new ArrayList<String>(keys.length);
                ArrayList<Response<String>> responses = new ArrayList<Response<String>>(keys.length);
                ExtendedShardedJedisPipeline pipeline = new ExtendedShardedJedisPipeline(jedis);
                for (int i = 0; i < keys.length; i++) {
                    responses.add(pipeline.set(keys[i], SerializationUtils.serialize(os[i])));
                }
                pipeline.sync();
                for (Response response : responses) {
                    ts.add((String) response.get());
                }
                return ts;
            }
        }.operate();
    }

    public String set(final String key, final Object o, final int expireTime) throws IOException {
        return new CountOperationFactory<String>(MAX_CACHE_OPERATE_COUNT, key) {
            public String cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                return jedis.setex(SafeEncoder.encode(key), expireTime, SerializationUtils.serialize(o));
            }
        }.operate();
    }

    public Long setnx(String key, final Object o) throws IOException {
        return new CountOperationFactory<Long>(MAX_CACHE_OPERATE_COUNT, key) {
            public Long cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                return jedis.setnx(SafeEncoder.encode(key), SerializationUtils.serialize(o));
            }
        }.operate();
    }

    public void delete(final String key) throws IOException {
        new CountOperationFactory<Long>(MAX_CACHE_OPERATE_COUNT, key) {
            public Long cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                return jedis.getShard(key).del(SafeEncoder.encode(key));
            }
        }.operate();
    }

    public <T> T get(final String key) throws IOException {
        try {
            return new CountOperationFactory<T>(MAX_CACHE_OPERATE_COUNT, key) {
                public T cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                    byte[] value = jedis.get(SafeEncoder.encode(key));
                    return (T) SerializationUtils.deserialize(value);
                }
            }.operate();
        } catch (JedisException e) {
            LOGGER.error("exception occur when get a object, key: " + key, e);
            return null;
        }
    }

    public <T> List<T> get(String[] keys) throws IOException {
        try {
            return new CountOperationFactory<List<T>>(MAX_CACHE_OPERATE_COUNT, keys) {
                public List<T> cacheOperate(ShardedJedis jedis) throws IOException {
                    List<T> ts = new ArrayList<T>(keys.length);
                    List<Response<byte[]>> responses = new ArrayList<Response<byte[]>>(keys.length);
                    ExtendedShardedJedisPipeline pipeline = new ExtendedShardedJedisPipeline(jedis);
                    for (String key : keys) {
                        responses.add(pipeline.getByte(key));
                    }
                    pipeline.sync();
                    for (Response response : responses) {
                        T t = (T) SerializationUtils.deserialize((byte[]) response.get());
                        ts.add(t);
                    }
                    return ts;
                }
            }.operate();
        } catch (JedisException e) {
            LOGGER.error("exception occur when get a object, keys: " + keys, e);
            return null;
        }
    }

    public Long incrby(String key, long increment) throws IOException {
        throw new UnsupportedOperationException();
    }

    public long zAdd(String key, final double score, final Object o) throws IOException {
        return new CountOperationFactory<Long>(MAX_CACHE_OPERATE_COUNT, key) {
            public Long cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                return jedis.zadd(SafeEncoder.encode(key), score, SerializationUtils.serialize(o));
            }
        }.operate();
    }

    /**
     * 插入长度限制
     *
     * @param key
     * @param score
     * @param o
     * @param len
     * @return
     * @throws IOException
     */
    public long zAdd(String key, double score, Object o, long len) throws IOException {
        throw new UnsupportedOperationException();
    }

    public long zAdd(String key, Map<Double, Object> map, long len) throws IOException {
        throw new UnsupportedOperationException();
    }

    public long zAdd(String key, Map<Double, Object> map) throws IOException {
        throw new UnsupportedOperationException();
    }

    public long zCard(String key) throws IOException {
        throw new UnsupportedOperationException();
    }

    public Double zScore(String key, Object o) throws IOException {
        throw new UnsupportedOperationException();
    }

    public long zCount(String key, double min, double max) throws IOException {
        throw new UnsupportedOperationException();
    }

    public long zRemByValue(String key, final Object value) throws IOException {
        return new CountOperationFactory<Long>(MAX_CACHE_OPERATE_COUNT, key) {
            public Long cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                return jedis.zrem(SafeEncoder.encode(key), SerializationUtils.serialize(value));
            }
        }.operate();
    }

    public long zRemrangeByRank(String key, final int start, final int end) throws IOException {
        return new CountOperationFactory<Long>(MAX_CACHE_OPERATE_COUNT, key) {
            public Long cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                return jedis.zremrangeByRank(SafeEncoder.encode(key), start, end);
            }
        }.operate();
    }

    public long zRemrangeByScore(String key, final double start, final double end) throws IOException {
        return new CountOperationFactory<Long>(MAX_CACHE_OPERATE_COUNT, key) {
            public Long cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                return jedis.zremrangeByScore(SafeEncoder.encode(key), start, end);
            }
        }.operate();
    }

    public <T> Set<T> zRevRange(String key, int start, int end) throws IOException {
        throw new UnsupportedOperationException();
    }

    public <T> Set<T> zRevRangeByOffset(String key, int start, int end) throws IOException {
        throw new UnsupportedOperationException();
    }

    public Set<Tuple> zRevWithScores(String key, int start, int end) throws IOException {
        throw new UnsupportedOperationException();
    }

    public <T> Set<T> zRangeByScore(String key, double min, double max, boolean rev) throws IOException {
        throw new UnsupportedOperationException();
    }

    public <T> Set<T> zRangeByScore(String key, double min, double max, boolean rev, int offset, int length) throws IOException {
        throw new UnsupportedOperationException();
    }

    public Set<Tuple> zRangeByScoreWithScores(String key, double min, double max, boolean rev, int offset, int length) throws IOException {
        throw new UnsupportedOperationException();
    }

    public Void hmSet(String[] keys, final Object[] os, final int expireTime) throws IOException {
        if (keys.length != os.length || keys.length == 0) {
            return null;
        }
        final List<Map<byte[], byte[]>> maps = new ArrayList<Map<byte[], byte[]>>();
        for (Object o : os) {
            if (o != null) {
                maps.add(parseMapFromObject(o));
            }
        }
        return new CountOperationFactory<Void>(MAX_CACHE_OPERATE_COUNT, keys) {
            public Void cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                ExtendedShardedJedisPipeline pipeline = new ExtendedShardedJedisPipeline(jedis);
                int validCount = 0;
                for (int i = 0; i < keys.length; i++) {
                    if (os[i] != null) {
                        pipeline.hmsetByte(keys[i], maps.get(validCount++));
                        if (expireTime > 0) {
                            pipeline.expire(keys[i], expireTime);
                        }
                    }
                }
                pipeline.sync();
                return null;
            }
        }.operate();
    }

    public Void hmSet(String[] keys, Object[] os) throws IOException {
        return hmSet(keys, os, -1);
    }

    public Void hmSet(String key, Object o, final int expireTime) throws IOException {
        final Map<byte[], byte[]> map = parseMapFromObject(o);
        return new CountOperationFactory<Void>(MAX_CACHE_OPERATE_COUNT, key) {
            public Void cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                ExtendedShardedJedisPipeline pipeline = new ExtendedShardedJedisPipeline(jedis);
                pipeline.hmsetByte(key, map);
                if (expireTime > 0) {
                    pipeline.expire(key, expireTime);
                }
                pipeline.sync();
                return null;
            }
        }.operate();
    }

    public String hmSet(String key, Object o) throws IOException {
        final Map<byte[], byte[]> map = parseMapFromObject(o);
        return new CountOperationFactory<String>(MAX_CACHE_OPERATE_COUNT, key) {
            public String cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                return jedis.hmset(SafeEncoder.encode(key), map);
            }
        }.operate();
    }

    public void hmSet(String key, final Map<String, Object> map) throws IOException {
        new CountOperationFactory<String>(MAX_CACHE_OPERATE_COUNT, key) {
            public String cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                if (map == null || map.isEmpty()) {
                    return null;
                }
                Map<byte[], byte[]> binaryMap = new HashMap<byte[], byte[]>(map.size());
                for (Map.Entry entry : map.entrySet()) {
                    binaryMap.put(SerializationUtils.serialize(entry.getKey()), SerializationUtils.serialize(entry.getValue()));
                }
                return jedis.hmset(SafeEncoder.encode(key), binaryMap);
            }
        }.operate();
    }

    /**
     * 只序列化某些字段
     *
     * @param key
     * @param o
     * @param fieldNameSet
     * @throws IOException
     */
    public void hmSet(String key, Object o, Set<String> fieldNameSet) throws IOException {
        final Map<byte[], byte[]> map = parseMapFromObject(o, fieldNameSet);
        new CountOperationFactory<String>(MAX_CACHE_OPERATE_COUNT, key) {
            public String cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                return jedis.hmset(SafeEncoder.encode(key), map);
            }
        }.operate();
    }

    public void hmSetCreateKeyIfNotExists(String key, Map<String, Object> map) throws IOException {
        throw new UnsupportedOperationException();
    }

    public long hSet(String key, final String field, final String value) throws IOException {
        return new CountOperationFactory<Long>(MAX_CACHE_OPERATE_COUNT, key) {
            public Long cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                return jedis.hset(SafeEncoder.encode(key), SerializationUtils.serialize(field),
                        SerializationUtils.serialize(value));
            }
        }.operate();
    }

    public long hSetCreateKeyIfNotExists(String key, String field, String value) throws IOException {
        throw new UnsupportedOperationException();
    }

    public String hGet(String key, final String field) throws IOException {
        return new CountOperationFactory<String>(MAX_CACHE_OPERATE_COUNT, key) {
            public String cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                return (String) SerializationUtils.deserialize(jedis.hget(SafeEncoder.encode(key),
                        SerializationUtils.serialize(field)));
            }
        }.operate();
    }

    public Object hmGet(final String key, final Class clazz) throws IOException {
        return new CountOperationFactory<Object>(MAX_CACHE_OPERATE_COUNT, key) {
            public Object cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                Map<byte[], byte[]> map = jedis.hgetAll(SafeEncoder.encode(key));
                return parseMap2Object(map, clazz);
            }
        }.operate();
    }

    public long hDel(String key) throws IOException {
        return new CountOperationFactory<Long>(MAX_CACHE_OPERATE_COUNT, key) {
            public Long cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                byte[] byteKey = SafeEncoder.encode(key);
                Set<byte[]> keySet = jedis.hkeys(byteKey);
                if (CollectionUtils.isEmpty(keySet)) {
                    return 0L;
                }
                return jedis.getShard(key).hdel(byteKey, keySet.toArray(new byte[][]{}));
            }
        }.operate();
    }

    public Long hIncrBy(String key, String field, int increment) throws IOException {
        throw new UnsupportedOperationException();
    }

    public Long hIncrBy(String key, String field, int increment, long defaultValue) throws JedisException, IOException {
        throw new UnsupportedOperationException();
    }

    public boolean hExists(String key) throws IOException {
        throw new UnsupportedOperationException();
    }

    public Long listLength(String key) throws IOException {
        throw new UnsupportedOperationException();
    }

    public <T> long listAppend(String key, T[] objs) throws IOException {
        throw new UnsupportedOperationException();
    }

    public <T> long listAppend(String key, T objs) throws IOException {
        throw new UnsupportedOperationException();
    }

    public <T> List<T> listRange(String key, int start, int end) throws IOException {
        throw new UnsupportedOperationException();
    }

    public boolean expire(String key, int seconds) throws IOException {
        throw new UnsupportedOperationException();
    }

    public boolean exists(String key) throws IOException {
        throw new UnsupportedOperationException();
    }

    public <T> void listDel(String key, T[] objs) throws IOException {
        throw new UnsupportedOperationException();
    }

    public <T> void listDel(String key, T obj) throws IOException {
        throw new UnsupportedOperationException();
    }

    public void hmSetMap(String key, Map<String, String> map) throws IOException {
        throw new UnsupportedOperationException();
    }

    public Map<String, String> hmGetMap(String key) throws IOException {
        throw new UnsupportedOperationException();
    }

    public <T> List<T> hmGet(String[] keys, final Class clazz) throws IOException {
        try {
            return new CountOperationFactory<List<T>>(MAX_CACHE_OPERATE_COUNT, keys) {
                public List<T> cacheOperate(ShardedJedis jedis) throws JedisException, IOException {
                    ExtendedShardedJedisPipeline pipeline = new ExtendedShardedJedisPipeline(jedis);
                    List<T> rl = new ArrayList<T>(keys.length);
                    List<Response<Map<byte[], byte[]>>> responses = new ArrayList<Response<Map<byte[], byte[]>>>(keys.length);
                    for (String key : keys) {
                        responses.add(pipeline.hgetBytesAll(key));
                    }
                    pipeline.sync();
                    Map<byte[], byte[]> all;
                    for (Response response : responses) {
                        all = (Map<byte[], byte[]>) response.get();
                        rl.add((T) parseMap2Object(all, clazz));
                    }
                    return rl;
                }
            }.operate();
        } catch (JedisException e) {
            LOGGER.error("exception occur when calling hmGetMaps, keys: " + keys, e);
            return null;
        }
    }

    public <T> T rPop(String key) throws IOException {
        throw new UnsupportedOperationException();
    }

    public Double zIncrby(String key, long increment, Object member) throws IOException {
        throw new UnsupportedOperationException();
    }

    public long zAddWithMap(String key, Map<Object, Double> map, long len) throws IOException {
        throw new UnsupportedOperationException();
    }

    /**
     * 将Bean对象序列化存储
     *
     * @param o
     * @return
     * @throws IOException
     */
    protected Map<byte[], byte[]> parseMapFromObject(Object o) throws IOException {
        if (o == null) {
            return null;
        }
        Map<byte[], byte[]> map = new HashMap<byte[], byte[]>();
        Field[] fields = o.getClass().getDeclaredFields();
        try {
            try {
                for (Field field : fields) {
                    boolean isStatic = Modifier.isStatic(field.getModifiers());
                    if (field != null && field.getName() != null && !isStatic) {
                        field.setAccessible(true);
                        Object obj = field.get(o);
                        if (obj == null) {
                            continue;
                        }
                        if (obj instanceof BigDecimal) {
                            map.put(SerializationUtils.serialize(field.getName()), SerializationUtils.serialize(obj));
                        } else {
                            String value = beanUtilsBean.getProperty(o, field.getName());
                            map.put(SerializationUtils.serialize(field.getName()), SerializationUtils.serialize(value));
                        }
                    }
                }
            } catch (InvocationTargetException e) {
                LOGGER.error("Can't parse Object to Map", e);
            } catch (NoSuchMethodException e) {
                LOGGER.error("Can't parse Object to Map", e);
            }
        } catch (IllegalAccessException e) {
            LOGGER.error("Can't parse Object to Map", e);
        }
        return map;
    }

    /**
     * 指定对象字段序列化
     *
     * @param o
     * @param fieldNameSet
     * @return
     */
    protected Map<byte[], byte[]> parseMapFromObject(Object o, Set<String> fieldNameSet) {
        if (o == null || CollectionUtils.isEmpty(fieldNameSet)) {
            return null;
        }
        Field[] fields = o.getClass().getDeclaredFields();
        Map<byte[], byte[]> map = new HashMap<byte[], byte[]>();
        try {
            for (Field field : fields) {
                boolean isStatic = Modifier.isStatic(field.getModifiers());
                if (field == null || field.getName() == null || isStatic || !fieldNameSet.contains(field.getName())) {
                    continue;
                }
                Object obj = field.get(o);
                if (obj instanceof BigDecimal) {
                    map.put(SerializationUtils.serialize(field.getName()), SerializationUtils.serialize(obj));
                } else {
                    String value = beanUtilsBean.getProperty(o, field.getName());
                    map.put(SerializationUtils.serialize(field.getName()), SerializationUtils.serialize(value));
                }
            }
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (NoSuchMethodException e) {
            e.printStackTrace();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        }
        return map;
    }


    /**
     * 对象反序列化
     *
     * @param map
     * @param Clazz
     * @return
     * @throws IOException
     */
    // TODO: 2016/5/31 Class缓存队列优化
    protected Object parseMap2Object(Map<byte[], byte[]> map, Class<?> Clazz) throws IOException {
        Object o;
        if (map == null || map.size() == 0) {
            return null;
        }
        Field[] fields = Clazz.getDeclaredFields();
        Map<String, Class<?>> fieldMap = new HashMap<String, Class<?>>();
        for (Field field : fields) {
            fieldMap.put(field.getName(), field.getType());
        }
        try {
            o = Clazz.newInstance();
            Map<String, Object> valueMap = new HashMap<String, Object>();
            for (Map.Entry<byte[], byte[]> entry : map.entrySet()) {
                String field = (String) SerializationUtils.deserialize(entry.getKey());
                Object obj = SerializationUtils.deserialize(entry.getValue());
                if (obj == null || field == null) {
                    continue;
                }
                obj = TypeConverter.convertObject(fieldMap.get(field), obj);
                valueMap.put(field, obj);
            }
            BeanMap beanMap = BeanMap.create(o);
            beanMap.setBean(o);
            beanMap.putAll(valueMap);
            return o;
        } catch (InstantiationException e) {
            LOGGER.error("Can't parse Object from List", e);
        } catch (IllegalAccessException e) {
            LOGGER.error("Can't parse Object from List", e);
        }
        return null;
    }

    /**
     * 缓存命令执行工厂
     *
     * @param <T>
     */
    protected abstract class CountOperationFactory<T> {
        int count;
        String key; //暂时用于报错时提供信息
        String[] keys; //暂时没用


        public CountOperationFactory(int count, String key) {
            this.count = count;
            this.key = key;
        }

        public CountOperationFactory(int count, String[] keys) {
            this.count = count;
            this.keys = keys;
        }

        public T operate() throws JedisException, IOException {
            if (count <= 0) {
                throw new JedisException("cache operate count less than 1");
            }
            while (count > 0) {
                try {
                    return jedisTemplate();
                } catch (JedisException e) {
                    LOGGER.error("cache CountOperation exception, ERROR :" + e);
                    count--;
                }
            }
            throw new JedisException("cache CountOperation exception ");
        }

        private T jedisTemplate() throws JedisException, IOException {
            ShardedJedis jedis = null;
            boolean isFailed = false;
            try {
                jedis = redisDataSource.getRedisClient();
                return cacheOperate(jedis);
            } catch (JedisException e) {
                isFailed = true;
                if (jedis != null && key != null) {
                    JedisShardInfo shardInfo = jedis.getShardInfo(key);
                    LOGGER.error("JedisException@{}:{}", shardInfo.getHost(), shardInfo.getPort());
                }
                throw e;
            } finally {
                if (jedis != null) {
                    redisDataSource.returnResource(jedis, isFailed);
                } else {
                    redisDataSource.returnResource(jedis);
                }
            }
        }

        public abstract T cacheOperate(ShardedJedis jedis) throws JedisException, IOException;
    }

}
